

package wal

import (
	"bufio"
	"encoding/binary"
	"hash"
	"io"
	"sync"

	"go.etcd.io/etcd/pkg/v3/crc"
	"go.etcd.io/etcd/pkg/v3/pbutil"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/storage/wal/walpb"
)

const minSectorSize = 512

// frameSizeBytes是以字节为单位的帧大小，包括记录大小和填充大小。
const frameSizeBytes = 8

type decoder struct {
	mu  sync.Mutex	// 在decoder开始读取日志文件时，需要加锁同步。
	// 该decoder实例通过该字段中记录的Reader实例读取相应的日志文件，
	// 这些日志文件就是在wal.openAtlndex（）方法中打开的日志文件。
	brs []*bufio.Reader
	// 读取日志记录的指针。
	lastValidOff int64
	// 校验码。
	crc          hash.Hash32
}

func newDecoder(r ...io.Reader) *decoder {
	readers := make([]*bufio.Reader, len(r))
	for i := range r {
		readers[i] = bufio.NewReader(r[i])
	}
	return &decoder{
		brs: readers,
		crc: crc.New(0, crcTable),
	}
}

// 实际读取WAL日志的功能是在decoder.decode（）的方法中完成
func (d *decoder) decode(rec *walpb.Record) error {
	rec.Reset()
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.decodeRecord(rec)
}

// etcd服务器中raft最大消息大小设置为1 MB 
// 假设项目设置了合理的消息大小限制，
// 因此条目大小不得超过10 MB 
const maxWALEntrySizeLimit = int64(10 * 1024 * 1024)

func (d *decoder) decodeRecord(rec *walpb.Record) error {
	// 检测brs字段长度，决定是否还有日志文件需要读取（
	if len(d.brs) == 0 {
		return io.EOF
	}
	// 读取第一个日志文件中的第一个日志记录的长度
	l, err := readInt64(d.brs[0])
	// 是否读到文件尾，或是读取到了预分配的部分，这都表示读取操作结束
	if err == io.EOF || (err == nil && l == 0) {
		d.brs = d.brs[1:]	// 更新brs字段，将其中第一个日志文件对应的Reader清除掉
		if len(d.brs) == 0 {	// 如果后面没有其他日志文件可读则返回EOF异常，表示读取正常结束
			return io.EOF
		}
		d.lastValidOff = 0	// 若后续还有其他日志文件待读取，则需要切换文件，这里会重置lastValidOff
		return d.decodeRecord(rec)	// 递归调用decodeRecord（）方法
	}
	if err != nil {
		return err
	}

	recBytes, padBytes := decodeFrameSize(l)
	if recBytes >= maxWALEntrySizeLimit-padBytes {
		return ErrMaxWALEntrySizeLimitExceeded
	}
	// 计算当前日志记录的实际长度及填无数据的长度，并创建相应的data切片
	data := make([]byte, recBytes+padBytes)
	// 从日志文件中读取指定长度的字节数如果读取不到指定的字节数，则会返回EOF异常，
	// 此时认为读取到了写入一半的日志记录，需要返回ErrUnexpectedEOF异常
	if _, err = io.ReadFull(d.brs[0], data); err != nil {
		if err == io.EOF {
			err = io.ErrUnexpectedEOF
		}
		return err
	}
	// 将O~recBytes反序列化成Record
	if err := rec.Unmarshal(data[:recBytes]); err != nil {
		if d.isTornEntry(data) {
			return io.ErrUnexpectedEOF
		}
		return err
	}

	// 进行crc校验
	if rec.Type != crcType {
		d.crc.Write(rec.Data)
		if err := rec.Validate(d.crc.Sum32()); err != nil {
			if d.isTornEntry(data) {
				return io.ErrUnexpectedEOF
			}
			return err
		}
	}
	// 将lastValidOff后移，准备读取下一条日志记录
	d.lastValidOff += frameSizeBytes + recBytes + padBytes
	return nil
}

func decodeFrameSize(lenField int64) (recBytes int64, padBytes int64) {
	// 记录大小存储在64位长度的低位56位
	recBytes = int64(uint64(lenField) & ^(uint64(0xff) << 56))
	// 非零填充由set MSb指示/负长度
	if lenField < 0 {
		// 填充存储在低位3位MSb 
		padBytes = int64((uint64(lenField) >> 56) & 0x7)
	}
	return recBytes, padBytes
}

// IStory确定最后一个条目《华尔街日报》的部分内容是
// 由于一段文字被撕破而损坏。
func (d *decoder) isTornEntry(data []byte) bool {
	if len(d.brs) != 1 {
		return false
	}

	fileOff := d.lastValidOff + frameSizeBytes
	curOff := 0
	chunks := [][]byte{}
	// 分割扇区边界上的数据
	for curOff < len(data) {
		chunkLen := int(minSectorSize - (fileOff % minSectorSize))
		if chunkLen > len(data)-curOff {
			chunkLen = len(data) - curOff
		}
		chunks = append(chunks, data[curOff:curOff+chunkLen])
		fileOff += int64(chunkLen)
		curOff += chunkLen
	}

	// 如果某个扇区块的任何数据都是0，那么它就是一个已损坏的写入
	for _, sect := range chunks {
		isZero := true
		for _, v := range sect {
			if v != 0 {
				isZero = false
				break
			}
		}
		if isZero {
			return true
		}
	}
	return false
}

func (d *decoder) updateCRC(prevCrc uint32) {
	d.crc = crc.New(prevCrc, crcTable)
}

func (d *decoder) lastCRC() uint32 {
	return d.crc.Sum32()
}

func (d *decoder) lastOffset() int64 { return d.lastValidOff }

func mustUnmarshalEntry(d []byte) raftpb.Entry {
	var e raftpb.Entry
	pbutil.MustUnmarshal(&e, d)
	return e
}

func mustUnmarshalState(d []byte) raftpb.HardState {
	var s raftpb.HardState
	pbutil.MustUnmarshal(&s, d)
	return s
}

func readInt64(r io.Reader) (int64, error) {
	var n int64
	err := binary.Read(r, binary.LittleEndian, &n)
	return n, err
}
